use super::*;
use crate::utils::SANDBOX_CONFIG;
use bytes::Bytes;
use chrono::TimeZone;
use dashmap::{DashMap, DashSet};
use eztrade_core::exchange::{Exchange, ExchangeTrait};
use eztrade_dto::{
    order,
    stage::StageInfo,
    events::{self, DataType, SubscribeEvent, GlobalEvent, Frequency},
};
use futures::future::select_all;
use rayon::prelude::*;
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::LazyLock;
use std::sync::{Arc, Mutex as StdMutex};
use tokio::sync::{Mutex, broadcast, mpsc};
use tokio::time::{Duration, Interval};
use tracing::{Instrument, instrument};

pub struct SimpleTimeSimulator {
    start_time_fake: DateTime<Utc>,
    prev_time_fake: StdMutex<DateTime<Utc>>,
    /// 真实开始时间, 用于计算模拟时间
    start_time_real: DateTime<Utc>,
    scale: f64,
    ticker: Mutex<Interval>,
}

/// 简易的时间模拟器
impl SimpleTimeSimulator {
    pub fn from_duration(tick_duration: Duration) -> Self {
        let now = Utc::now();
        SimpleTimeSimulator {
            start_time_fake: now,
            prev_time_fake: StdMutex::new(now),
            start_time_real: now,
            scale: SANDBOX_CONFIG.time_scale,
            ticker: Mutex::new(tokio::time::interval(tick_duration)),
        }
    }
}

impl TimeWarpTrait for SimpleTimeSimulator {
    fn align_with(mut self, start_time: DateTime<Utc>) -> Self {
        self.start_time_real = Utc::now();
        self.start_time_fake = start_time;
        self.prev_time_fake = StdMutex::new(start_time);
        self
    }

    fn scale_with(mut self, scale: f64) -> Self {
        self.scale = scale;
        self
    }

    fn get_simulated_time(&self) -> DateTime<Utc> {
        let time_delta = Utc::now() - self.start_time_real;
        let scaled_millisec = time_delta.num_milliseconds() as f64 * self.scale;
        let dtime_fake = chrono::Duration::milliseconds(scaled_millisec as i64);
        self.start_time_fake + dtime_fake
    }

    fn get_simulated_elapsed(&self) -> chrono::Duration {
        let fake_now = self.get_simulated_time();
        let mut prev_time_fake = self.prev_time_fake.lock().unwrap();
        let v = fake_now - *prev_time_fake;
        *prev_time_fake = fake_now;
        v
    }
}

/// Simple SandBox Simulator
///
/// 基于 SubscribeEvent 内置通道的简化设计
pub struct SimpleSandboxSimulator {
    stage_info: Arc<RwLock<StageInfo>>,
    time_warp: Arc<SimpleTimeSimulator>,
    main_task_handle: StdMutex<Option<tokio::task::JoinHandle<()>>>,
    transfer_task_handle: StdMutex<Option<tokio::task::JoinHandle<()>>>,

    // 交易所引擎
    exchange: Arc<Exchange>,

    // 订阅事件管理: 所有的 SubscribeEvent 实例
    subscribe_events: Arc<DashSet<Arc<SubscribeEvent>>>,

    // 客户端通道管理: client_id -> broadcast sender
    client_senders: Arc<DashMap<u64, broadcast::Sender<Message>>>,

    // 全局事件管理: 直接管理 GlobalEvent 实例
    global_events: Arc<DashSet<Arc<GlobalEvent>>>,
}

impl SandboxTrait for SimpleSandboxSimulator {
    fn new(stage: StageInfo) -> Self {
        let time_unit = stage.time_unit_us;
        let time_warp = match &stage.market_sim {
            Some(market_sim) => {
                let scale = market_sim.time_multiplier;
                let tick_duration = Duration::from_micros(time_unit / scale as u64);
                let time_start =
                    DateTime::<Utc>::from_timestamp_micros(market_sim.simulate_start_us).unwrap();
                let time_warp = SimpleTimeSimulator::from_duration(tick_duration)
                    .scale_with(scale)
                    .align_with(time_start);
                Arc::new(time_warp)
            }
            None => {
                let tick_duration = Duration::from_micros(time_unit);
                Arc::new(SimpleTimeSimulator::from_duration(tick_duration))
            }
        };

        let exchange = Arc::new(Exchange::default());

        SimpleSandboxSimulator {
            stage_info: Arc::new(RwLock::new(stage)),
            time_warp,
            main_task_handle: StdMutex::new(None),
            transfer_task_handle: StdMutex::new(None),
            exchange,
            subscribe_events: Arc::new(DashSet::new()),
            client_senders: Arc::new(DashMap::new()),
            global_events: Arc::new(DashSet::new()),
        }
    }

    fn get_stage_phase(&self) -> stage::StagePhase {
        self.stage_info.read().unwrap().phase.clone()
    }

    fn get_stage_info(&self) -> StageInfo {
        self.stage_info.read().unwrap().clone()
    }

    fn start(&self) -> Result<()> {
        let (stage_id, stage_name) = {
            let info = self.stage_info.read().unwrap();
            (info.id, info.name.clone())
        };
        tracing::info!("Starting sandbox stage: {}", stage_name);

        // 检查是否已经启动了任务
        {
            let task_guard = self.main_task_handle.lock().unwrap();
            if task_guard.is_some() {
                tracing::warn!("Sandbox stage {} is already started", stage_name);
                return Ok(());
            }
        }

        // 启动撮合引擎
        let exchange = self.exchange.clone();
        let _match_handler = tokio::spawn(exchange.run(SANDBOX_CONFIG.tick_us));

        // 启动数据搬运线程
        let subscribe_events = self.subscribe_events.clone();
        let client_senders = self.client_senders.clone();
        let global_events = self.global_events.clone();
        
        let transfer_handle = tokio::spawn(async move {
            loop {
                // 每次循环重新收集所有receiver和对应信息
                let mut futures: Vec<std::pin::Pin<Box<dyn std::future::Future<Output = (Option<Message>, Option<u64>, usize, bool)> + Send>>> = Vec::new();
                
                // 收集个人订阅events的receiver
                for (idx, subscribe_event) in subscribe_events.iter().enumerate() {
                    let receiver = subscribe_event.get_receiver();
                    let client_id = subscribe_event.client_id;
                    
                    futures.push(Box::pin(async move {
                        let mut guard = receiver.lock().await;
                        (guard.recv().await, Some(client_id), idx, false) // false表示个人订阅
                    }));
                }
                
                // 收集全局events的receiver
                for (idx, global_event) in global_events.iter().enumerate() {
                    let receiver = global_event.get_receiver();
                    
                    futures.push(Box::pin(async move {
                        let mut guard = receiver.lock().await;
                        (guard.recv().await, None, idx, true) // true表示全局事件
                    }));
                }
                
                // 如果没有任何receiver，等待一段时间后继续
                if futures.is_empty() {
                    tokio::time::sleep(Duration::from_millis(100)).await;
                    continue;
                }

                // 使用select_all等待任何一个消息->(result, index, remaining)
                let (result, _, remaining_futures) = select_all(futures).await;
                
                // 立即释放未完成的futures以避免内存泄漏
                drop(remaining_futures);
                
                match result {
                    (Some(message), client_id_opt, _idx, is_global) => {
                        if is_global {
                            // 全局事件：广播给所有客户端
                            for client_sender_entry in client_senders.iter() {
                                let client_sender = client_sender_entry.value();
                                let _ = client_sender.send(message.clone());
                            }
                        } else {
                            // 个人订阅：发送给特定客户端
                            if let Some(client_id) = client_id_opt {
                                if let Some(client_sender) = client_senders.get(&client_id) {
                                    let _ = client_sender.send(message);
                                }
                            }
                        }
                    }
                    (None, client_id_opt, idx, is_global) => {
                        // Receiver断开，记录警告
                        // 断开的receiver会在下次循环时被自动排除（如果对应的event被移除）
                        if is_global {
                            tracing::warn!("Global event receiver disconnected at index {}", idx);
                        } else {
                            tracing::warn!("Subscribe event receiver disconnected for client {:?} at index {}", client_id_opt, idx);
                        }
                    }
                }
            }
        });

        // 启动数据生成任务
        let time_warp = self.time_warp.clone();
        let subscribe_events_for_gen = self.subscribe_events.clone();
        let global_events_for_gen = self.global_events.clone();
        let exchange = self.exchange.clone();

        let generate_handle = tokio::spawn(async move {
            // 获取当前有订阅者的数据类型和品种，添加 is_global 标记
            let mut active_types: HashSet<(DataType, Option<String>, Option<Frequency>, bool)> = HashSet::new();
            loop {
                // 使用时间扭曲的 ticker
                {
                    let mut ticker_warp = time_warp.ticker.lock().await;
                    ticker_warp.tick().await;
                }
                
                // 使用模拟时间更新撮合引擎时间戳
                exchange.update_timestamp(time_warp.get_simulated_time().timestamp_micros() as u64);

                // 从全局事件中收集（标记为全局）
                for global_event in global_events_for_gen.iter() {
                    active_types.insert((global_event.dtype.clone(), global_event.symbol.clone(), global_event.frequency.clone(), true));
                }
                
                // 从具体订阅事件中收集（标记为个人）
                for subscribe_event in subscribe_events_for_gen.iter() {
                    active_types.insert((subscribe_event.dtype.clone(), subscribe_event.symbol.clone(), subscribe_event.frequency.clone(), false));
                }

                // 为每个活跃的数据类型生成模拟数据
                active_types.par_iter().for_each(|(dtype, symbol, frequency, is_global)| {
                    let simulate_time = time_warp.get_simulated_time();
                    
                    // 只处理 Tick 数据类型
                    if *dtype == DataType::Tick {
                        if let Some(symbol_str) = symbol {
                            let mock_data = match exchange.tick_data(symbol_str) {
                                Ok(tick_data) => {
                                    // 如果有tick数据，直接使用
                                    Some(Message::TickData(tick_data))
                                }
                                _ => {
                                    // 如果没有tick数据，生成模拟订单并插入进撮合引擎
                                    let sys_orders = Self::generate_mock_order(
                                        dtype,
                                        symbol,
                                    );
                                    for order in sys_orders {
                                        if let Err(e) = exchange.place_order(order, 0) {
                                            tracing::error!("Failed to place mock order: {}", e);
                                        }
                                    }
                                    None
                                }
                            };

                            // 如果有数据，根据 is_global 标记快速判断发送目标
                            if let Some(message) = mock_data {
                                if *is_global {
                                    // 发送给全局事件
                                    for global_event in global_events_for_gen.iter() {
                                        if global_event.dtype == *dtype 
                                            && global_event.symbol == *symbol 
                                            && global_event.frequency == *frequency {
                                            let _ = global_event.send(message.clone());
                                        }
                                    }
                                } else {
                                    // 发送给个人订阅事件
                                    for subscribe_event in subscribe_events_for_gen.iter() {
                                        if subscribe_event.dtype == *dtype 
                                            && subscribe_event.symbol == *symbol 
                                            && subscribe_event.frequency == *frequency {
                                            let _ = subscribe_event.send(message.clone());
                                        }
                                    }
                                }
                            }
                        }
                    } else {
                        // TODO: 处理其他数据类型
                        match dtype {
                            DataType::OrderDone => {
                                
                            }
                            _ => {}
                        }
                    }
                });
            }
        });

        // 保存任务句柄
        {
            let mut task_guard = self.main_task_handle.lock().unwrap();
            *task_guard = Some(generate_handle);
        }
        {
            let mut transfer_guard = self.transfer_task_handle.lock().unwrap();
            *transfer_guard = Some(transfer_handle);
        }

        Ok(())
    }

    fn subscribe(&self, client_id: u64, subscribe: SubscribeEvent) -> Result<()> {
        // 1. 确保客户端有发送器
        if !self.client_senders.contains_key(&client_id) {
            let (client_sender, _) = broadcast::channel::<Message>(1000);
            self.client_senders.insert(client_id, client_sender);
        }

        // 2. 创建订阅事件并添加到订阅管理中
        let subscribe_arc = Arc::new(subscribe);
        self.subscribe_events.insert(subscribe_arc.clone());

        // 3. 将订阅事件与组件关联
        self.exchange.client_subscribe(client_id, subscribe_arc.clone())?;

        Ok(())
    }

    fn unsubscribe(&self, client_id: u64, subscribe: SubscribeEvent) -> Result<()> {
        // 查找并移除匹配的订阅事件
        let subscribe_arc = Arc::new(subscribe);
        self.subscribe_events.remove(&subscribe_arc);

        Ok(())
    }

    fn subscribe_global(&self, global_event: GlobalEvent) -> Result<()> {
        // 将 GlobalEvent 实例添加到管理中
        let global_event_arc = Arc::new(global_event);
        self.global_events.insert(global_event_arc.clone());

        Ok(())
    }

    fn unsubscribe_global(&self, global_event: GlobalEvent) -> Result<()> {
        // 移除全局事件实例
        self.global_events.remove(&global_event);

        Ok(())
    }

    fn get_receiver(&self, client_id: u64) -> Result<Receiver<Message>> {
        // 确保客户端有发送器
        if !self.client_senders.contains_key(&client_id) {
            let (client_sender, _) = broadcast::channel::<Message>(1000);
            self.client_senders.insert(client_id, client_sender);
        }
        log::debug!("Creating receiver for client {}", client_id);

        // 获取客户端的接收器
        if let Some(client_sender) = self.client_senders.get(&client_id) {
            Ok(client_sender.subscribe())
        } else {
            Err(anyhow::anyhow!(
                "Failed to create receiver for client {}",
                client_id
            ))
        }
    }

    #[tracing::instrument(skip(self, message))]
    fn process_message(&self, client_id: u64, message: Message) -> Option<Message> {
        match message {
            Message::Notification(msg) => {
                tracing::trace!("Received notification message: {:?}", msg);
                // 告知
                let response = Bytes::from("Notification received");
                None
            }
            Message::Order(order) => { // 下单
                tracing::trace!("Received order message: {:?}", order);
                // 使用撮合引擎处理订单
                match self.exchange.place_order(order, client_id) {
                    Ok(placed_order) => {
                        tracing::trace!("Order placed successfully: {:?}", placed_order);
                        let response = Message::Order(placed_order);
                        return Some(response);
                    }
                    Err(e) => {
                        tracing::error!("Failed to place order: {}", e);
                        None
                    }
                }
            }
            Message::QueryAccount() => { // 查询账户
                match self.exchange.get_portfolio(client_id){
                    Ok(portfolio) => {
                        tracing::trace!("Retrieved portfolio: {:?}", portfolio);
                        let response = Message::Portfolio(portfolio);
                        return Some(response);
                    }
                    Err(e) => {
                        tracing::error!("Failed to retrieve portfolio: {}", e);
                        None
                    }
                }
            }
            Message::OrderOperation(operation) => { // 操作账户
                match operation.operation {
                    order::Operation::Cancel => {
                        self.exchange.cancel_order(operation.id, client_id).unwrap_or_else(|e| {
                            tracing::error!("Failed to cancel order: {}", e);
                        });
                        None
                    }
                    order::Operation::Check => {
                        match self.exchange.query_order(operation.id, client_id) {
                            Some(order) => {
                                tracing::trace!("Retrieved order: {:?}", order);
                                Some(Message::Order(order))
                            }
                            None => {
                                tracing::error!("Failed to retrieve order: {}", operation.id);
                                None
                            }
                        }
                    }
                }
            }
            _ => {
                tracing::warn!("不是该被服务侧处理的消息类型: {:?}", message);
                None
            }
        }
    }
}

impl SimpleSandboxSimulator {
    /// 停止沙盒任务
    pub fn stop(&self) -> Result<()> {
        let mut task_guard = self.main_task_handle.lock().unwrap();
        if let Some(handle) = task_guard.take() {
            handle.abort();
        }
        Ok(())
    }

    /// 清理指定客户端的所有订阅
    pub fn cleanup_client(&self, client_id: u64) -> Result<()> {
        // 收集所有该客户端的订阅事件
        let events_to_remove: Vec<_> = self.subscribe_events
            .iter()
            .filter(|event| event.client_id == client_id)
            .map(|event| event.clone())
            .collect();

        // 移除所有该客户端的订阅
        for event in events_to_remove {
            self.subscribe_events.remove(&event);
        }

        // 移除客户端发送器
        self.client_senders.remove(&client_id);

        Ok(())
    }

    /// 为指定的数据类型和品种生成模拟订单
    fn generate_mock_order(
        dtype: &DataType,
        symbol: &Option<String>,
    ) -> Vec<order::Order> {
        let mid = 5000.0; // 模拟最后成交价
        let vol = 5000;

        let symbol_str = match symbol {
            Some(s) => s.clone(),
            None => "Unknown".to_string(),
        };

        let mut orders = Vec::new();
        // 生成买单
        let mut buy_order = order::Order::default();
        buy_order.symbol = symbol_str.clone();
        buy_order.direction = order::Direction::Long;
        buy_order.position_effect = order::PositionEffect::Open;
        buy_order.price = Some(mid - 0.01);
        buy_order.quantity = vol;
        orders.push(buy_order);

        // 生成卖单
        let mut sell_order = order::Order::default();
        sell_order.symbol = symbol_str.clone();
        sell_order.direction = order::Direction::Short;
        sell_order.position_effect = order::PositionEffect::Open;
        sell_order.price = Some(mid + 0.01);
        sell_order.quantity = 2 * vol;
        orders.push(sell_order);

        orders
    }
}
